//
// Copyright 2011 Aster Data

import java.io.File;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.List;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.SimpleAnalyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.store.SimpleFSLockFactory;
import org.apache.lucene.util.Version;

import com.asterdata.ncluster.sqlmr.ArgumentClause;
import com.asterdata.ncluster.sqlmr.ClientVisibleException;
import com.asterdata.ncluster.sqlmr.OutputInfo;
import com.asterdata.ncluster.sqlmr.RowFunction;
import com.asterdata.ncluster.sqlmr.RuntimeContract;
import com.asterdata.ncluster.sqlmr.TaskInfo;
import com.asterdata.ncluster.sqlmr.data.ColumnDefinition;
import com.asterdata.ncluster.sqlmr.data.RowEmitter;
import com.asterdata.ncluster.sqlmr.data.RowIterator;

public final class lucene_index implements RowFunction {
    //
    // These member variables will be populated with the values
    // of the argument clauses passed to your SQL-MR function.
    //

    private String indexdirArgument = "";
    private String analyzerArgument = "";
    private String overwriteindexArgument = "";
    private String textcolumnnameArgument = "";
    private int textcolumnIndex = -1;
    private String docidcolumnnameArgument = "";
    private int docidcolumnIndex = -1;
    private int taskindex;

    //
    // The constructor establishes the RuntimeContract between
    // your function and nCluster. During query planning, your
    // function will constructed on a single node. During query
    // execution, it will be constructed and run on one or more
    // nodes.
    //

    public lucene_index(RuntimeContract contract) {

        //
        // Construct the output schema.
        //

        List<ColumnDefinition> outputColumns = new ArrayList<ColumnDefinition>();

        outputColumns.addAll(contract.getInputInfo().getColumns());

        //
        // Read argument clauses into appropriate member variables.
        //

        ArgumentClause tmpAc;
        String tmpValue;

        //
        // Read argument clause 'indexdir'
        //

        tmpAc = contract.useArgumentClause("indexdir");

        tmpValue = tmpAc.getSingleValue();
        indexdirArgument = tmpValue;

        // optional argument overwriteindex
        if (contract.hasArgumentClause("overwriteindex"))
            overwriteindexArgument = contract.useArgumentClause("overwriteindex").getSingleValue();

        //
        // Read argument clause 'analyzer'
        //

        tmpAc = contract.useArgumentClause("analyzer");

        tmpValue = tmpAc.getSingleValue();
        analyzerArgument = tmpValue;

        //
        // Read argument clause 'textcolumnname'
        //

        tmpAc = contract.useArgumentClause("textcolumnname");

        tmpValue = tmpAc.getSingleValue();
        textcolumnIndex = contract.getInputInfo().getColumnIndex(tmpValue);

        //
        // Read argument clause 'docidcolumnname'
        //

        tmpAc = contract.useArgumentClause("docidcolumnname");

        tmpValue = tmpAc.getSingleValue();
        docidcolumnIndex = contract.getInputInfo().getColumnIndex(tmpValue);

        //
        // Add any other function-specific initialization of member variables
        // or other checks here. Throw a ClientVisibleException to signal
        // an error to the client.
        //

        //
        // Complete the contract
        //

        contract.setOutputInfo(new OutputInfo(outputColumns));
        contract.complete();
    }

    public void operateOnSomeRows(RowIterator inputIterator, RowEmitter outputEmitter) {

        // **************
        // todo - where to persist the text index? cannot create dir on /primary/tmp. /tmp is ok
        // how to make it participate in replication. does taskindex change on failover?
        // **************

        taskindex = TaskInfo.getTaskIndex();
        File indexdir = new File(indexdirArgument + "/" + taskindex);

        // initialize analyze
        Analyzer analyzer = null;
        if (analyzerArgument.equals("simple"))
            analyzer = new SimpleAnalyzer();
        else if (analyzerArgument.equals("standard"))
            analyzer = new StandardAnalyzer(Version.LUCENE_30);
        else
            throw new ClientVisibleException("analyzer parameter value \"" + analyzerArgument
                    + "\" is not valid. supported values are simple/standard.");

        // initialize lucene
        try {

            Directory dir = FSDirectory.open(indexdir, new SimpleFSLockFactory(indexdir));//
            IndexWriter writer = null;
            if (!overwriteindexArgument.toLowerCase().equals("true")) { // do not overwrite existing index, append to
                                                                        // existing index
                writer = new IndexWriter(dir, analyzer, false, IndexWriter.MaxFieldLength.UNLIMITED);
            } else { // overwrite existing index
                writer = new IndexWriter(dir, analyzer, true, IndexWriter.MaxFieldLength.UNLIMITED);
            }
            writer.setMergeFactor(100);

            // index each row
            Document document = null;
            while (inputIterator.advanceToNextRow()) {
                document = new Document();
                document.add(new Field("text", analyzer.tokenStream("text",
                        new StringReader(inputIterator.getStringAt(textcolumnIndex)))));
                document.add(new Field("docid", inputIterator.getStringAt(docidcolumnIndex), Field.Store.YES,
                        Field.Index.NOT_ANALYZED_NO_NORMS));
                writer.addDocument(document);
            }

            writer.optimize();
            writer.close();
            dir.close();

        } catch (Exception e) {
            StackTraceElement[] strace = e.getStackTrace();
            String msg = "";
            for (int i = 0; i < strace.length; i++)
                msg += strace[i].toString() + "\n";
            throw new ClientVisibleException("lucene indexing exception: " + e.toString() + "\n" + msg);
        }
    }

}
